package sf.r2dbc.sql;

import io.r2dbc.spi.ColumnMetadata;
import io.r2dbc.spi.Connection;
import io.r2dbc.spi.Result;
import io.r2dbc.spi.Row;
import io.r2dbc.spi.RowMetadata;
import io.r2dbc.spi.Statement;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import sf.common.exception.SmallOrmException;
import sf.common.log.LogContext;
import sf.common.log.OrmLog;
import sf.common.wrapper.Page;
import sf.database.dao.DBContext;
import sf.database.dialect.DBDialect;
import sf.database.jdbc.sql.PageStrategy;
import sf.database.jdbc.sql.SQLContext;
import sf.database.jdbc.sql.SQLParameter;
import sf.database.util.DBUtils;
import sf.r2dbc.binding.BindMarker;
import sf.r2dbc.binding.BindMarkersFactory;
import sf.r2dbc.handle.R2dbcPageListHandler;
import sf.r2dbc.rowmapper.R2dbcMapRowMapper;
import sf.r2dbc.rowmapper.R2dbcRowMapper;
import sf.r2dbc.rowmapper.R2dbcRowMapperHelp;
import sf.spring.util.Assert;
import sf.spring.util.CollectionUtils;
import sf.spring.util.LinkedCaseInsensitiveMap;
import sf.tools.NumberUtils;
import sf.tools.StringUtils;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.function.BiFunction;
import java.util.function.Function;

public class R2dbcCrudSqlImpl implements R2dbcCrudSqlInf {
    private static final Logger LOGGER = LoggerFactory.getLogger(R2dbcCrudSqlImpl.class);

    private static R2dbcCrudSqlImpl instance = new R2dbcCrudSqlImpl();

    private R2dbcCrudSqlImpl() {

    }

    public static R2dbcCrudSqlImpl getInstance() {
        return instance;
    }

    /**
     * @param connection
     * @param sql
     * @param paras
     * @return
     * @see org.springframework.r2dbc.core.ColumnMapRowMapper
     */
    @Override
    public Flux<Map<String, Object>> select(Connection connection, String sql, List<SQLParameter> paras) {
        return select(connection, R2dbcMapRowMapper.INSTANCE, sql, paras);
    }

    @Override
    public <T> Mono<T> selectOne(Connection connection, Class<T> beanClass, String sql, List<SQLParameter> parameters) {
        SQLContext sqlContext = new SQLContext();
        sqlContext.setSql(sql);
        sqlContext.setParas(parameters);
        return R2dbcCrud.getInstance().getCrudModel().selectOne(connection, beanClass, sqlContext);
    }

    @Override
    public <T> Flux<T> selectList(Connection connection, Class<T> beanClass, long start, int limit, String sql, List<SQLParameter> parameters) {
        DBDialect dialect = R2dbcUtils.getDBDialect(connection);
        String tempSql = dialect.sqlPageList(new StringBuilder(sql), start, limit).toString();
        SQLContext sqlContext = new SQLContext();
        sqlContext.setSql(tempSql);
        sqlContext.setParas(parameters);
        return R2dbcCrudModelImpl.getInstance().selectList(connection, beanClass, sqlContext);
    }

    @Override
    public <T> Flux<T> selectList(Connection connection, Class<T> beanClass, String sql, List<SQLParameter> parameters) {
        SQLContext sqlContext = new SQLContext();
        sqlContext.setSql(sql);
        sqlContext.setParas(parameters);
        return R2dbcCrudModelImpl.getInstance().selectList(connection, beanClass, sqlContext);
    }

    @Override
    public <T> Mono<Page<T>> selectPage(Connection conn, long start, int limit, Class<T> beanClass, String sql,
                                        Object... parameters) {
        Assert.notNull(beanClass, "beanClass cannot be null.");
        Assert.notNull(sql, "sql cannot be null.");
        String countSql = DBUtils.getSmartSqlSelectCount(sql);
        return selectPageRaw(conn, start, limit, beanClass, countSql, parameters, sql, parameters);
    }

    @Override
    public <T> Mono<Page<T>> selectPageRaw(Connection conn, long start, int limit, Class<T> beanClass, String countSql, Object[] countParas, String listSql, Object[] listParas) {
        return selectPageRaw(conn, start, limit, beanClass, countSql, countParas, listSql, listParas, PageStrategy.onlyList);
    }

    @Override
    public <T> Mono<Page<T>> selectPageRaw(Connection conn, long start, int limit, Class<T> beanClass, String countSql, Object[] countParas, String listSql, Object[] listParas, PageStrategy strategy) {
        Assert.notNull(beanClass, "beanClass cannot be null.");
        Assert.hasText(countSql, "count sql cannot be null,please check you sql");
        Assert.hasText(listSql, "list sql cannot be null,please check you sql");
        Assert.notNull(strategy, "strategy cannot be null.");

        return selectOne(conn, Long.class, countSql, countParas).defaultIfEmpty(0L).flatMap(count -> {
            Mono<List<T>> items = Mono.just(Collections.emptyList());
            if (count > 0) {
                switch (strategy) {
                    case onlyList://只查询列表,不包含分页参数.
                        String pageSql = R2dbcUtils.getDBDialect(conn).sqlPageList(new StringBuilder(listSql), start, limit).toString();
                        if (StringUtils.isNotBlank(pageSql)) {
                            items = selectList(conn, beanClass, pageSql, listParas).collectList();// 使用数据库自身的分页SQL语句，将直接返回某一个
                        } else {
                            throw new SmallOrmException("not find page sql,please check you sql");
                        }
                        break;
                    case fake: //假分页 如果不支持分页，那么使用原始的分页方法
                        DBContext dbContext = R2dbcUtils.getDBContext(conn);
                        R2dbcRowMapper<T> rowMapper = R2dbcRowMapperHelp.getRowMapper(beanClass, dbContext);
                        R2dbcPageListHandler<T> rsh = new R2dbcPageListHandler<>(rowMapper);
                        rsh.setFirstResult((int) start);
                        rsh.setMaxResults(limit);
                        items = selectResult(conn, rsh, listSql, listParas).collectList();
                        break;
                    case hasOffsetLimit: // 查询语句已包含offset,limit 使用数据库自身的分页SQL语句，将直接返回结果
                        items = selectList(conn, beanClass, listSql, listParas).collectList();
                        break;
                    default:
                        break;
                }
            }
            return items.map(it -> {
                Page<T> page = new Page<>(start, limit, count);
                page.setList(it);
                return page;
            });
        });
    }

    @Override
    public <T> Flux<T> selectResult(Connection conn, R2dbcResultSetCallback<T> callback, String sql, Object... parameters) {
        Assert.notNull(callback, "callback cannot be null.");
        Assert.notNull(sql, "sql cannot be null.");

        LogContext log = OrmLog.commonLog(null, sql, parameters);

        BindMarkersFactory bmf = R2dbcUtils.getBindMarkersFactory(conn);
        List<BindMarker> bindMarkerList = new ArrayList<>();
        String innerSql = R2dbcUtils.parseSql(bmf, bindMarkerList, sql);

        Statement statement = conn.createStatement(innerSql);
        R2dbcSql.fillSQLStatement(statement, bindMarkerList, parameters);

        OrmLog.sqlLog(null, log, () -> R2dbcUtils.getAutoCommit(conn));
        return callback.callback(Flux.from(statement.execute()))
//                .doOnNext(OrmLog::resultOnly)
                .doOnComplete(() -> OrmLog.resultTime(null, log))
                .doOnCancel(() -> OrmLog.resultTime(null, log))
                .doOnError(throwable -> OrmLog.resultLog(null, log, null));
    }

    public <T> Flux<T> select(Connection conn, BiFunction<Row, RowMetadata, T> mappingFunction, String sql, List<SQLParameter> parameters) {
        Assert.notNull(mappingFunction, "rsh cannot be null.");
        Assert.notNull(sql, "sql cannot be null.");

        LogContext log = OrmLog.commonLog(null, sql, parameters);

        BindMarkersFactory bmf = R2dbcUtils.getBindMarkersFactory(conn);
        List<BindMarker> bindMarkerList = new ArrayList<>();
        String innerSql = R2dbcUtils.parseSql(bmf, bindMarkerList, sql);

        Statement statement = conn.createStatement(innerSql);
        R2dbcSql.fillSQLStatement(statement, bindMarkerList, parameters);

        OrmLog.sqlLog(null, log, () -> R2dbcUtils.getAutoCommit(conn));
        return Flux.from(statement.execute())
                .flatMap(r -> r.map(mappingFunction))
//                .doOnNext(OrmLog::resultOnly)
                .doOnComplete(() -> OrmLog.resultTime(null, log))
                .doOnCancel(() -> OrmLog.resultTime(null, log))
                .doOnError(throwable -> OrmLog.resultLog(null, log, null));
    }

    public <T> Flux<T> select(Connection conn, BiFunction<Row, RowMetadata, T> mappingFunction, String sql, Object... parameters) {
        Assert.notNull(mappingFunction, "mappingFunction cannot be null.");
        Assert.notNull(sql, "sql cannot be null.");

        LogContext log = OrmLog.commonLog(null, sql, parameters);

        BindMarkersFactory bmf = R2dbcUtils.getBindMarkersFactory(conn);
        List<BindMarker> bindMarkerList = new ArrayList<>();
        String innerSql = R2dbcUtils.parseSql(bmf, bindMarkerList, sql);

        Statement statement = conn.createStatement(innerSql);
        R2dbcSql.fillSQLStatement(statement, bindMarkerList, parameters);

        OrmLog.sqlLog(null, log, () -> R2dbcUtils.getAutoCommit(conn));
        return Flux.from(statement.execute())
                .flatMap(r -> r.map(mappingFunction))
//                .doOnNext(OrmLog::resultOnly)
                .doOnComplete(() -> OrmLog.resultTime(null, log))
                .doOnCancel(() -> OrmLog.resultTime(null, log))
                .doOnError(throwable -> OrmLog.resultLog(null, log, null));
    }

    @Override
    public Mono<Integer> insert(Connection connection, List<String> pkeys, Map<String, Object> keyValues, String sql, List<SQLParameter> parameters) {

        LogContext log = new LogContext();
        log.setStart(System.currentTimeMillis());
        log.setSql(sql);

        BindMarkersFactory bmf = R2dbcUtils.getBindMarkersFactory(connection);
        List<BindMarker> bindMarkerList = new ArrayList<>();
        String innerSql = R2dbcUtils.parseSql(bmf, bindMarkerList, sql);

        Statement statement = connection.createStatement(innerSql);
        R2dbcSql.fillSQLStatement(statement, bindMarkerList, parameters);

//        OrmLog.sqlLog(null, log, () -> R2dbcUtils.getAutoCommit(connection));
        if (CollectionUtils.isNotEmpty(pkeys)) {
            statement = statement.returnGeneratedValues(pkeys.toArray(new String[]{}));
            return Flux.from(statement.execute()).flatMap(r -> r.map((row, metadata) -> {
                        for (String columnName : pkeys) {
                            Object o = row.get(columnName);
                            keyValues.put(columnName, o);
                        }
                        return 1;
                    })) //
                    .last(1)
                    .doOnNext(c -> OrmLog.resultSqlLog(null, log, c, () -> R2dbcUtils.getAutoCommit(connection)))
                    .doOnError(throwable -> OrmLog.resultSqlLog(null, log, (Exception) throwable));
        } else {
            return Flux.from(statement.execute()).flatMap(Result::getRowsUpdated).cast(Number.class)
                    .map(NumberUtils::numberToInt).last(0)
                    .doOnNext(c -> OrmLog.resultSqlLog(null, log, c, () -> R2dbcUtils.getAutoCommit(connection)))
                    .doOnError(throwable -> OrmLog.resultSqlLog(null, log, (Exception) throwable));
        }
    }

    @Override
    public Mono<Integer> execute(Connection connection, String sql, List<SQLParameter> paras) {
        SQLContext sqlContext = new SQLContext();
        sqlContext.setParas(paras);

        LogContext log = OrmLog.commonLog(null, sql, sqlContext.getValues());

        BindMarkersFactory bmf = R2dbcUtils.getBindMarkersFactory(connection);
        List<BindMarker> bindMarkerList = new ArrayList<>();
        String innerSql = R2dbcUtils.parseSql(bmf, bindMarkerList, sql);

        Statement statement = connection.createStatement(innerSql);

        R2dbcSql.fillSQLStatement(statement, bindMarkerList, paras);
//        OrmLog.sqlLog(null, log, () -> R2dbcUtils.getAutoCommit(connection));
        return Flux.from(statement.execute()).flatMap(Result::getRowsUpdated).cast(Number.class)
                .map(NumberUtils::numberToInt).last(0)
                .doOnNext(c -> OrmLog.resultSqlLog(null, log, c, () -> R2dbcUtils.getAutoCommit(connection)))
                .doOnError(throwable -> OrmLog.resultSqlLog(null, log, null, () -> R2dbcUtils.getAutoCommit(connection), (Exception) throwable));
    }


    /**
     * @param connection
     * @param sql
     * @param parameters
     * @return
     * @see org.springframework.r2dbc.core.ColumnMapRowMapper
     */
    @Override
    public Flux<Map<String, Object>> select(Connection connection, String sql, Object... parameters) {
        return select(connection, R2dbcMapRowMapper.INSTANCE, sql, parameters);
    }

    @Override
    public <T> Mono<T> selectOne(Connection connection, Class<T> beanClass, String sql, Object... parameters) {
        Assert.notNull(beanClass, "beanClass cannot be null.");
        DBContext dbContext = R2dbcUtils.getDBContext(connection);
        R2dbcRowMapper<T> rsh = R2dbcRowMapperHelp.getRowMapper(beanClass, dbContext);
        return select(connection, rsh, sql, parameters).next();
    }

    @Override
    public <T> Flux<T> selectList(Connection connection, Class<T> beanClass, long start, int limit, String sql, Object... parameters) {
        Assert.notNull(beanClass, "beanClass cannot be null.");
        DBContext dbContext = R2dbcUtils.getDBContext(connection);
        DBDialect dialect = R2dbcUtils.getDBDialect(connection);
        sql = dialect.sqlPageList(new StringBuilder(sql), start, limit).toString();
        R2dbcRowMapper<T> rsh = R2dbcRowMapperHelp.getRowMapper(beanClass, dbContext);
        return select(connection, rsh, sql, parameters);
    }

    @Override
    public <T> Flux<T> selectList(Connection connection, Class<T> beanClass, String sql, Object... parameters) {
        Assert.notNull(beanClass, "beanClass cannot be null.");
        DBContext dbContext = R2dbcUtils.getDBContext(connection);
        R2dbcRowMapper<T> rsh = R2dbcRowMapperHelp.getRowMapper(beanClass, dbContext);
        return select(connection, rsh, sql, parameters);
    }

    @Override
    public <T> T execute(Connection con, Function<Connection, T> action) {
        Assert.notNull(action, "Callback object must not be null");
        return action.apply(con);
    }

    @Override
    public Mono<Integer> execute(Connection connection, String sql, Object... parameters) {

        LogContext log = OrmLog.commonLog(null, sql, parameters);

        BindMarkersFactory bmf = R2dbcUtils.getBindMarkersFactory(connection);
        List<BindMarker> bindMarkerList = new ArrayList<>();
        String innerSql = R2dbcUtils.parseSql(bmf, bindMarkerList, sql);

        Statement statement = connection.createStatement(innerSql);

        R2dbcSql.fillSQLStatement(statement, bindMarkerList, parameters);
//        OrmLog.sqlLog(null, log, () -> R2dbcUtils.getAutoCommit(connection));
        return Flux.from(statement.execute()).flatMap(Result::getRowsUpdated).cast(Number.class)
                .map(NumberUtils::numberToInt).last(0)
                .doOnNext(c -> OrmLog.resultSqlLog(null, log, c, () -> R2dbcUtils.getAutoCommit(connection)))
                .doOnError(throwable -> OrmLog.resultSqlLog(null, log, throwable));
    }

    @Override
    public Mono<Integer> execute(Connection connection, String sql) {
        Statement statement = connection.createStatement(sql);
        return Mono.from(statement.execute()).flatMapMany(Result::getRowsUpdated).cast(Number.class)
                .map(NumberUtils::numberToInt).last(0);
    }

    @Override
    public Mono<Integer> insertFast(Connection conn,
                                    String sql, Object... parameters) {
        return execute(conn, sql, parameters);
    }

    @Override
    public Mono<Integer> insert(Connection conn, List<String> pkeys, Map<String, Object> keyValues, String sql, Object... parameters) {
        return execute(conn, sql, parameters, false, pkeys, keyValues);
    }

    @Override
    public Mono<Integer> execute(Connection conn, String sql, Object[] parameters, boolean insertFast,
                                 List<String> pkeys, Map<String, Object> keyValues) {
        List<Object[]> objects = Collections.singletonList(parameters);
        List<Map<String, Object>> mapList = new ArrayList<>(2);
        return executeBatch(conn, sql, objects, false, pkeys, mapList).last(0).flatMap(count -> {
            if (keyValues != null && !mapList.isEmpty()) {
                keyValues.putAll(mapList.get(0));
            }
            return Mono.just(count);
        });
    }

    @Override
    public Flux<Integer> executeBatch(Connection conn, String sql, List<Object[]> parameters) {
        return executeBatch(conn, sql, parameters, true, null, null);
    }

    @Override
    public Flux<Integer> executeBatch(Connection conn, String sql, List<Object[]> parameters, boolean insertFast,
                                      List<String> pkeys, List<Map<String, Object>> keyValues) {
        Assert.notNull(sql, "sql cannot be null.");
//        Assert.notNull(parameters, "parameters is null.");

        int size = parameters == null ? 0 : parameters.size();
        boolean isSingle = size <= 1;
        boolean isInsert = false;
        boolean returnKey = false;
        if (StringUtils.containsIgnoreCase(sql, "insert")) {
            isInsert = true;
        }
        DBDialect dialect = R2dbcUtils.getDBDialect(conn);
        LogContext log = new LogContext();
        if (size == 0) {
            OrmLog.commonLog(log, sql, parameters);
        } else {
            log.setStart(System.currentTimeMillis());
            log.setSql(sql);
        }

        if (CollectionUtils.isNotEmpty(pkeys) && isInsert && !insertFast && keyValues != null && !dialect.isNosql()) {
            returnKey = true;
        }
        BindMarkersFactory bmf = R2dbcUtils.getBindMarkersFactory(conn);
        List<BindMarker> bindMarkerList = new ArrayList<>();
        String innerSql = R2dbcUtils.parseSql(bmf, bindMarkerList, sql);

        Statement statement = conn.createStatement(innerSql);

        if (CollectionUtils.isNotEmpty(parameters)) {
            int count = 0;
            boolean first = true;
            for (Object[] parameter : parameters) {
                OrmLog.batchCommonLog(null, log, sql, !isSingle, Integer.MAX_VALUE, (count++) + 1, parameter);
                if (!first) {
                    statement.add();
                }
                R2dbcSql.fillSQLStatement(statement, bindMarkerList, parameter);
                first = false;
            }
        }
        if (OrmLog.needShowSql()) {
            log.setResult(new ArrayList<>(size == 0 ? 1 : size));
        }
        if (returnKey) {
            statement = statement.returnGeneratedValues(pkeys.toArray(new String[0]));
            return Flux.from(statement.execute()).flatMap(r -> r.map((row, metadata) -> {
                        Map<String, Object> result = new LinkedCaseInsensitiveMap<>(2);
                        for (ColumnMetadata cmd : metadata.getColumnMetadatas()) {
                            String columnName = cmd.getName();
                            result.put(columnName, R2dbcConvertUtils.get(row.get(columnName), null, null));
                        }
                        keyValues.add(result);
                        return 1;
                    }))
                    .doOnNext(t -> {
                        if (log.getResult() != null) {
                            ((List<Integer>) log.getResult()).add(t);
                        }
                    })
                    .doOnComplete(() -> OrmLog.resultSqlLog(null, log, log.getResult(), () -> R2dbcUtils.getAutoCommit(conn), null, true))
                    .doOnCancel(() -> OrmLog.resultSqlLog(null, log, log.getResult(), () -> R2dbcUtils.getAutoCommit(conn), null, true))
                    .doOnError(throwable -> OrmLog.resultSqlLog(null, log, throwable));
        } else {
            return Flux.from(statement.execute()).flatMap(Result::getRowsUpdated).cast(Number.class)
                    .map(NumberUtils::numberToInt)
                    .doOnNext(t -> {
                        if (log.getResult() != null) {
                            ((List<Integer>) log.getResult()).add(t);
                        }
                    })
                    .doOnComplete(() -> OrmLog.resultSqlLog(null, log, log.getResult(), () -> R2dbcUtils.getAutoCommit(conn), null, true))
                    .doOnCancel(() -> OrmLog.resultSqlLog(null, log, log.getResult(), () -> R2dbcUtils.getAutoCommit(conn), null, true))
                    .doOnError(throwable -> OrmLog.resultSqlLog(null, log, throwable));
        }
    }
}
